package kn.pulldata.config;


import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
public class RabbitMqConfig {

    @Autowired
    private ConnectionFactory connectionFactory;



    @Bean
    public RabbitTemplate rabbitTemplate() {

        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        // 设置 ConfirmCallback 确认消息是否到达交换机
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                System.out.println("Message with correlationId " + correlationData.getId() + " was acknowledged.");
            } else {
                // todo 保存数据库
                System.err.println("Message with correlationId " + correlationData.getId() + " failed: " + cause);
            }
        });

        // 设置 ReturnCallback 处理消息未到达队列时的回退
        rabbitTemplate.setReturnsCallback(returned -> {
            // todo 保存数据库
            System.err.println("Message returned: " + new String(returned.getMessage().getBody()) +
                    ", replyCode: " + returned.getReplyCode() +
                    ", replyText: " + returned.getReplyText() +
                    ", exchange: " + returned.getExchange() +
                    ", routingKey: " + returned.getRoutingKey());
        });

        return rabbitTemplate;
    }

    @Bean
    public TaskExecutor rabbitTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(40); // 初始线程数
        executor.setMaxPoolSize(40); // 最大线程数
        executor.setQueueCapacity(100); // 队列容量
        executor.setThreadNamePrefix("rabbitmq-thread-");
        executor.initialize();
        return executor;
    }
    @Bean
    public SimpleRabbitListenerContainerFactory factory1(ConnectionFactory connectionFactory, TaskExecutor rabbitTaskExecutor) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setTaskExecutor(rabbitTaskExecutor);
        factory.setConcurrentConsumers(10); // 初始并发消费
        factory.setMaxConcurrentConsumers(20); // 最大并发消费
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); // 手动确认
        return factory;
    }
}
